// -*-mode:c++; coding:utf-8-*-

#ifndef _XBASE_ROW_MUTATION_POOL_HPP_
#define _XBASE_ROW_MUTATION_POOL_HPP_

namespace xbase
{

	class row_mutation_pool
		: public async_module_base<io_service_pool::user_level,
					   io_service_pool::xbase_level>
	{
	public:
		typedef async_module_base<io_service_pool::user_level,
					  io_service_pool::xbase_level> base_type;
		typedef row_mutation_pool this_type;
		typedef boost::intrusive_ptr<this_type> ptr_type;
		typedef std::map<std::string, row_mutation_data_ptr> row_map;
		typedef std::map<row_mutation_data_ptr> row_vector;

		typedef boost::system::error_code error_code;
		typedef boost::asio::deadline_timer deadline_timer;

		enum apply_mode
		{
			single_row_mode,
			multi_row_mode
		};
		
	public:
		explicit
		row_mutation_pool(io_service_pool &pool,
				  location_cache &cache,
				  session_manager &manager,
				  protocol::table_info_ptr table_info)
			: base_type(pool, "row_mutation_pool")
			, m_session_manager(manager)
			, m_location_cache(cache)
			, m_timer(this->get_lower_level_service())
			, m_timer_actived(false)
			, m_table_info(table_info)
			, m_max_commit_count(10000)
			, m_max_row_count(1000)
			, m_max_wait_interval(9)
			, m_async_op_actived(false)
			, m_max_async_op_count(3) {
			active_timer(); // on lower level service
		}

		void commit_mutation(const row_mutation &mutation) {
			assert(mutation.table_name() == m_table_info->m_table_name);
			
			bool is_free_data_ptr = false;
			bool be_able_to_active = false;
			boost::mutex::scoped_lock lock(m_row_map_mutex);
			row_mutation_data_ptr &ptr = m_row_map[mutation.row_name()];
			if(! ptr) // 若原来不存在，则会新插入，那么指针为空
			{
				ptr = mutation.data_ptr();
			}
			else
			{
				ptr->add(mutation.data());
				is_free_data_ptr = true;
			}
			++m_commit_count;
			if((m_commit_count >= m_max_commit_count) ||
			   (m_row_map.size() >= m_max_row_count))
			{
				m_commit_count = 0;
				be_able_to_active = true;
			}
			lock.unlock();

			if(is_free_data_ptr)
			{
				boost::mutex::scoped_lock lock_free_data(m_free_mutation_data_mutex);
				row_mutation_data_ptr ptr = mutation.data_ptr();
				ptr->clear();
				m_free_mutation_data.push_back(ptr);
			}

			if(be_able_to_active)
			{
				active_async_commit();
			}
		}

		row_mutation get_mutation(const std::string &row_name) {
			assert(!row_name.empty());
			boost::mutex::scoped_lock lock(m_free_mutation_data_mutex);
			if(m_free_mutation_data.empty())
			{
				return row_mutation(m_table_info, row_name);
			}
			else
			{
				row_mutation_data_ptr ptr = m_free_mutation_data.front();
				m_free_mutation_data.pop_front();
				ptr->change_row_name(row_name);
				return row_mutation(m_table_info, ptr);
			}
		}

	protected:
		// 被用户接口调用
		void active_async_commit() {
			boost::mutex::scoped_lock lock(m_async_op_mutex);
			const bool be_able_to_active = ((! m_async_op_actived) &&
							true);
			//(m_pending_ops < m_max_pending_ops));
			if(be_able_to_active)
				m_async_op_actived = true;
			lock.unlock();
			if(be_able_to_active)
			{
				error_code ec;
				m_timer.cancel(ec);
				this->get_lower_level_service().
					dispatch(boost::bind(&this_type::async_commit,
							     this_ptr()));
			}
		}

		// 在构造函数中调用或在下层服务函数中调用，所以不必对 m_timer_actived 加锁
		void active_timer() {
			if(! m_timer_actived)
			{
				m_timer_actived = true;
				m_timer.expires_from_now(get_wait_interval());
				m_timer.async_wait(boost::bind(&this_type::on_timeout, this_ptr(), _1));
			}
		}

		boost::posix_time::seconds get_wait_interval() {
			return boost::posix_time::seconds(m_max_wait_interval);
		}

		ptr_type this_ptr() {
			return ptr_type(this);
		}

		// run in lower level service

		void async_commit_exit() {
			m_last_commit_null = false;
			boost::mutex::scoped_lock lock(m_async_op_mutex);
			m_async_op_actived = false;
			lock.unlock();
			active_timer();
		}

		// 流程：用上次提交的子表的结束行E，(lower_bound)找第一
		// 个大于等于E的行R，获取R所属的子表T，再查找所有属于T
		// 的行，将其移动出来提交，并更新结束行。若失败，则将未
		// 提交的行返回到row_map中。
		void async_commit() {
			if((! m_flush_mode) && // 在非flush模式下
			   m_last_commit_null) // 上次遍历没有行被写，则不再遍历
			{
				async_commit_exit();
				return;
			}

			boost::mutex::scoped_lock lock(m_row_map_mutex);
			if(m_row_map.empty())
			{
				lock.unlock();
				async_commit_exit();
				return;
			}

			m_row_map::iterator it;
			if(m_last_committed_tablet)
			{
				if(m_last_committed_tablet.m_end_row.empty()) // 上次是最后一个子表，则跳到最前
				{
					it = m_row_map.begin();
				}
				else
				{
					it = m_row_map.lower_bound(m_last_committed_tablet.m_end_row); // 上次是最后一个有数据的子表，也跳到最前
					if(it == m_row_map.end())
					{
						it = m_row_map.begin();
					}
				}
			}
			else	// 第一次提交操作，从第一行开始
			{
				it = m_row_map.begin();
			}
			if(it == m_row_map.begin())
			{
				m_some_rows_commited_per_cycle = false;
			}
			std::string start_row = *it;
			lock.unlock();

			m_location_cache.async_get_meta(boost::bind(&this_type::on_got_tablet_info,
								    this_ptr(),
								    _1, _2, _3),
							m_table_info->m_table_name,
							start_row,
							false,
							0);
		}

		// 取出属于该子表的所有行
		void on_got_tablet_info(protocol::tablet_info_ptr tablet_info,
					protocol::server_address_ptr address,
					const error_code &ec) {
			m_last_committed_tablet = tablet_info;
			
			bool touch_end = false;
			boost::mutex::scoped_lock lock(m_row_map_mutex);
			const row_map::iterator start = m_row_map.lower_bound(tablet_info->m_start_row);
			row_map::iterator end;
			if(tablet_info->m_end_row.empty()) // to end
			{
				end = m_row_map.end();
			}
			else
			{
				end = m_row_map.lower_bound(tablet_info->m_end_row);
			}

			if(end == m_row_map.end())
			{
				touch_end = true;
			}

			// no (enough) rows need commit for this tablet
			if((start == end) ||
			   ((! m_flush_mode) &&
			    (std::distance(start, end) < m_min_row_each_commit)))
			{
				LOG(DEBUG, "");
				async_commit(); // back to async_commit
			}

			row_vector_ptr rows(new row_vector);
			for(row_map::iterator it = start; it != end; ++it)
			{
				rows->push_back(it->second);
			}
			m_row_map.erase(start, end);
			lock.unlock();

			multi_row_mutator_ptr ptr;
			if(m_free_mutator.empty())
			{
				ptr.reset(new_async_module(multi_row_mutator, m_pool, m_session_manager, m_location_cache));
			}
			else
			{
				ptr = m_free_mutator.front();
				m_free_mutator.pop_front();
			}
			ptr->async_apply(boost::bind(&this_type::on_connected,
						     this_ptr(), rows,
						     _1, _2));
		}
		
		void on_complete(row_vector_ptr rows,
				 std::size_t success_rows, const error_code &ec) {
			if(ec)
			{
				if(success_rows < rows->size())
				{
					std::deque<row_mutation_data_ptr> free_data_queue;
					boost::mutex::scoped_lock lock_row_map(m_row_map_mutex);
					for(std::size_t i = success_rows; i < rows->size(); ++i)
					{
						const row_mutation_data_ptr &ptr_to_add = ((*rows)[i]);
						row_mutation_data_ptr &ptr = m_row_map[ptr_to_add->row_name()];
						if(! ptr) // 若原来不存在，则会新插入，那么指针为空
						{
							ptr = mutation.data_ptr();
						}
						else
						{
							ptr->add(*ptr_to_add);
							ptr_to_add->clear();
							free_data_queue.push_back(ptr_to_add);
						}
					}
					lock_row_map.unlock();
					boost::mutex::scoped_lock lock_free_data(m_free_mutation_data_mutex);
					m_free_mutation_data.insert(m_free_mutation_data.end(),
								    free_data_queue.begin(),
								    free_data_queue.end());
					lock_free_data.unlock();
				}
			}
			else if(success_rows != rows->size())
			{
				LOG(ERROR, "");
			}

			--m_pending_ops;
			if((m_pending_ops < m_max_pending_ops) &&
			   ((m_commit_count >= m_max_row_count) ||
			    (m_row_map.size() >= m_max_row_count) ||
			    (cur_time - m_commit_start_time >= m_max_wait_interval)))
			{
				active_async_commit();
			}
			else if(m_pending_ops == 0)
			{
				active_timer();
			}
		}

		void on_timeout(const error_code &ec) {
			if(! ec) // is timeout
			{
				boost::mutex::scoped_lock lock(m_row_map_mutex);
				bool be_able_to_active = (! m_row_map.empty());
				lock.unlock();
				
				if(be_able_to_active)
					async_commit(); // 最后一个异步提交操作激活计时器
				else
					active_timer(); // 再次等待
			}
			//else,	is cancelled
		}

	private:
		io_service_pool &m_pool;
		session_manager &m_manager;
		location_cache &m_location_cache;
		deadline_timer m_timer;
		bool m_timer_actived;
		protocol::table_info_ptr m_table_info;
	        bool m_stop;	// 如果为真，则计时器被取消后就不再启动了

		// options

		// 超过这些值将会触发异步提交操作
		const std::size_t m_max_commit_count;  // 最大提交次数
		const std::size_t m_max_row_count;     // 最大总行数
		const std::size_t m_max_wait_interval; // 最大等待时间
		
		const std::size_t m_max_pending_ops;  // 最多异步操作数

		std::size_t m_commit_count;
		std::size_t m_commit_start_time; // 提交启动时间
		std::size_t m_pending_ops;
		
		row_map m_row_map;
		boost::mutex m_row_map_mutex;

		protocol::tablet_info_ptr m_last_committed_tablet;
		bool m_some_rows_commited_per_cycle;
		bool m_last_commit_null;
		
		std::deque<row_mutation_data_ptr> m_free_mutation_data; // and is empty mutation data

		struct data
		{
			typedef std::vector<row_mutation_data_ptr> data_ptr_vector;
			
			boost::mutex m_mutex;
			data_ptr_vector m_data_vec;
			std::size_t m_num_completed;
			
			std::size_t m_num_succeeded;
		};
		typedef boost::shared_ptr<data> data_ptr;
		data_ptr m_committing_data;
		data_ptr m_failed_data;
		std::list<data_ptr> m_applying_data_list;
	};


} // namespace xbase

#endif	// _XBASE_ROW_MUTATION_POOL_HPP_
